--------------------------sql任务说明----------------------------------
-- @author :liuxiaojie05
-- @descrbe :话单b2c任务 此任务只sink出结束话单数据 ods --->dwd
-- @udf jar包 : bigdata_realtime_flink_sql_udf-1.0-testlxj-jar-with-dependencies.jar
-- 下游使用方：olap、dmp
--------------------------sql任务说明----------------------------------

-- 加载函数
CREATE FUNCTION ZpDWDB2CBillUdtf as 'udtf.etl.dwd.ZpDWDB2CBillUdtf';

-- source
CREATE TABLE zp_ods_b2c_bill (
	log  STRING
) with (
	'connector' = 'wmb',
	'topic' = '100657',
	'client.id' = '6',
	'properties.key' = 'F5993419EE58377033B8B7792814A71F',
	'properties.ip' = 'esb.registryserver.58dns.org',
	'properties.port' = '21010',
	'format' = 'raw'
);


-- sink
create table dwd_zp_b2c_bill
(
		appId     string,
        audioUrl  string,
        bindid    string,
        callType  string,
        callid    string,
        infoId    string,
        calltime  string,
        eSBKey    string,
        logType   string,
        releaseDir        string,
        requestId string,
        telA      string,
        telAUserId        string,
        telARole  string,
        telB      string,
        telBUserId        string,
        telBRole  string,
        telX      string,
        callStatus        string,
        bizCode   string,
        recordAddressOnWos string,
        hangup    string,
        record    string,
        tryDownloadTime   INT,
        ringTime  string,
        startTime string,
        stopTime  string,
        poolid    string,
        supplier  string,
        gwNumlocalCode    string,
        platform  string,
        localId   string,
        pageType  string,
        bindExpire        INT,
        validityBeginTime BIGINT,
        releaseCause      string,
        source    string,
        cateId    INT,
        extension string,
        mode      string,
        wosUrl    string,
        bizExt    string,
        bizGroup  string,
        groupId   string,
        extendCode        string,
        account   string,
        recordId  string,
        callRecording     string,
        controlType       string,
        callResult        string,
        calltimeout       string,
        sequenceId        string,
        sequence  string,
        buid      string,
        wmb_time  BIGINT,
        sink_time string,
        e_time   BIGINT
)
with(
'connector' = 'kafka',
'topic' = 'hdp_lbg_supin_dwd_zp_b2c_bill',
'properties.bootstrap.servers' = '10.178.10.8:9092,10.178.10.10:9092,10.178.10.11:9092,10.178.10.12:9092,10.178.10.13:9092',
'properties.group.id' = 'hdp_lbg_supin_dwd_zp_b2c_bill-prod-v1',
'scan.startup.mode' = 'group-offsets',
'properties.client.id' = 'hdp_lbg_supin-hdp_lbg_supin_dwd_zp_b2c_bill-LT2Qc',
'format' = 'json'
);

insert into dwd_zp_b2c_bill
select pp.* from zp_ods_b2c_bill, LATERAL TABLE(ZpDWDB2CBillUdtf(log)) as pp
-- 过滤出结束话单数据
where  CHAR_LENGTH(groupId) > 1  and CHAR_LENGTH(startTime) > 1 and CHAR_LENGTH(stopTime) > 1 and logType = 'RECORD'
;

